package ex.datastream.functions.richFunction;

import ex.vo.Card;
import ex.vo.Person;
import ex.vo.PersonCard;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.RichCoFlatMapFunction;
import org.apache.flink.util.Collector;

public class RichCoFlatMapFunc extends RichCoFlatMapFunction<Person, Card, PersonCard> {
    private ValueState<Person> personState;
    private ValueState<Card> cardState;

    @Override
    public void open(Configuration parameters) throws Exception {
        personState = getRuntimeContext().getState(
                new ValueStateDescriptor<Person>("personState", TypeInformation.of(new TypeHint<Person>() {
                })));
        cardState = getRuntimeContext().getState(new ValueStateDescriptor<Card>("cardState", TypeInformation.of(new TypeHint<Card>() {
        })));
    }

    @Override
    public void flatMap1(Person person, Collector<PersonCard> collector) throws Exception {
        Card card = cardState.value();
        //证件状态不为空:创建人证件对象，清空状态
        //证件状态为空，只更新人
        if (card != null) {
            collector.collect(new PersonCard(person, card));
            cardState.clear();
        } else {
            personState.update(person);
        }
    }

    @Override
    public void flatMap2(Card card, Collector<PersonCard> collector) throws Exception {
        //人状态不为空:创建人证件对象，清空状态
        //人状态为空，只更新证件
        Person person=personState.value();
        if(person!=null){
            collector.collect(new PersonCard(person,card));
            personState.clear();
        }else{
            cardState.update(card);
        }

    }
}
